﻿import os
import redis
import const
import time
import math
from MyUtil import print_error
import sqlite3
from const import HQ_DATA_FILE# 测试当日行情文件

# 这三个参数已经挪到main_feeddata.py
# HQ_DATA_FILE_TEST ='D:\\FBYX\\get_hq_x\\bigdata\HQ_DATA\\20180814hq - 副本.db' # 测试指定行情db文件
# symbol_test = '603706'
# SIMULATE = True  # 模拟
#FEED_TICK_NUM = 500
REP_TIME = 1    # '每个tick重复次数,模拟实盘网速快3秒内获取两个tick
STOP_SEND_CHANNEL = 'STOP'

class sqliteObj_feed(object):
    def __init__(self, HQ_DATA_FILE_TEST):
        try:
            self.conn = sqlite3.connect(HQ_DATA_FILE_TEST) #HQ_DATA_FILE
            self._cursor = self.conn.cursor()
            print('connection db sucess')
        except Exception as e:
            print_error(e)

    def close_db(self):
        print('close db')
        self._cursor.close()
        self.conn.close()

    def fetch_data_from_stock_table(self, tableName, N=600, all_tick=False):
        # '''查询所有数据...'''
        # print('查询一只股票所有数据...')
        if not all_tick:
            fetchall_sql = '''SELECT * FROM ''' + tableName + ''' limit ''' + str(N)
            return self.fetchall(self.conn, fetchall_sql)
        else:
            fetchall_sql = '''SELECT * FROM ''' + tableName
            return self.fetchall(self.conn, fetchall_sql)

    def fetch_data_from_stock_table_batch(self, tableName, time_begin, time_end):
        # '''查询所有数据...'''
        # print('查询一只股票所有数据...')
        fetchall_sql = '''SELECT * FROM ''' + tableName +' where time >=' + str(time_begin) + ' and time<'+str(time_end)

        return self.fetchall(self.conn, fetchall_sql)

    def fetchall(self, conn, sql):
        r = []
        if sql is not None and sql != '':
            cu = self._cursor
            # if True:
            #     print('执行sql:[{}]'.format(sql))
            cu.execute(sql)
            r = cu.fetchall()
            #print('len r:',len(r))
            """if len(r) > 0:
                for e in range(len(r)):
                    print(r[e])"""
        else:
            print('the [{}] is empty or equal None!'.format(sql))
        return r

    def get_table_names(self):
        # 获取表名
        table_sql = "select name from sqlite_master where type='table' order by name"
        self._cursor.execute(table_sql)
        table_names = self._cursor.fetchall()
        return table_names

    def readdata(self, symbol_test):
        data=self.fetch_data_from_stock_table('sh'+symbol_test)
        return data



class no_redis_main(object):
    def __init__(self, HQ_DATA_FILE_TEST, SIMULATE, symbol_test,FEED_TICK_NUM = 500, all_tick=True):
        self.tick_i = 0
        self.init_time = 9*3600 + 30*60
        self.data_dic = {}
        self.msqliteObj_feed = sqliteObj_feed(HQ_DATA_FILE_TEST)
        self.HQ_DATA_FILE_TEST = HQ_DATA_FILE_TEST
        print('db file :',HQ_DATA_FILE_TEST )
        self.end_time = 0
        self.batch_time_lst = [[93000, 100000],[100001,103000],[103001,110000],[110001,113000],
                               [130000,133000],[133001,140000],[140001,143000],[143001,145000]]
        if SIMULATE:
            self.table_names = self.msqliteObj_feed.get_table_names()
            print('begin send, table_names len:', len(self.table_names))
        # self.save_batch_lst = []
        # self.batch_idx = -1
        # self.save_batch_data()

    def save_batch_data(self):
        for item_batch in self.batch_time_lst:
            batch_lst= []
            for item in self.table_names:
                tb = 'sh' + item[0][2:]  # 固定sh开头,遗留bug
                symbol = item[0][2:]
                data_all = self.msqliteObj_feed.fetch_data_from_stock_table_batch(tb, item_batch[0], item_batch[1])
                print('data_all:',len(data_all))
                batch_lst.append(data_all)
            print('batch_lst:',len(batch_lst))
            self.save_batch_lst.append(batch_lst)

    def get_nex_batch(self):
        self.batch_idx += self.batch_idx
        self.data_dic.clear()
        self.data_dic = {}
        data_all = self.save_batch_lst[self.batch_idx]
        for item in self.table_names:
            tb = 'sh' + item[0][2:]  # 固定sh开头,遗留bug
            symbol = item[0][2:]
            if symbol not in self.data_dic:
                self.data_dic[symbol] = {}
            for data_item in data_all:
                line = getContents(data_item, symbol)[0]
                time_key = line['time']
                self.data_dic[symbol][time_key] = line
        try:
            print('read db batch !!:', sorted(self.data_dic['000001'].items())[0])
            print('read db batch !!:', sorted(self.data_dic['600000'].items())[0])
        except Exception as e:
            print_error(e)

    def get_batch_data(self, begin_time, end_time):
        print('get_batch_data:', begin_time, '-', end_time)

        self.data_dic.clear()
        self.data_dic = {}

        for item in self.table_names:
            tb = 'sh' + item[0][2:]  # 固定sh开头,遗留bug
            symbol = item[0][2:]
            data_all = self.msqliteObj_feed.fetch_data_from_stock_table_batch(tb, begin_time, end_time)
            if len(data_all) > 0:
                if symbol not in self.data_dic:
                    self.data_dic[symbol] = {}
                for data_item in data_all:
                    line = getContents(data_item, symbol)[0]
                    time_key = line['time']
                    self.data_dic[symbol][time_key] = line

        try:
            print('len data dic:', len(self.data_dic))
            if len(self.data_dic) > 0:
                print('read db batch !!:', sorted(self.data_dic['000001'].items())[0])
                print('read db batch !!:', sorted(self.data_dic['600000'].items())[0])
            else:
                print('!!!!! len data_dic is O')
        except Exception as e:
            print_error(e)
        # self.msqliteObj_feed.close_db()

    def get_tick(self):
        now_time_s = self.init_time + self.tick_i*3
        hours, minutes, seconds = self.get_hms(now_time_s)
        now_time_s_1 = self.init_time + self.tick_i * 3 - 1
        hours1, minutes1, seconds1 = self.get_hms(now_time_s_1)
        time_key_1 = hours1 + minutes1 + seconds1
        now_time_s_2 = self.init_time + self.tick_i * 3 - 2
        hours2, minutes2, seconds2 = self.get_hms(now_time_s_2)
        time_key_2 = hours2 + minutes2 + seconds2
        res = []    # 返回每3秒的每个对应股票
        ###
        if now_time_s > self.end_time:
            self.end_time = now_time_s + 60*30
            hours_e, minutes_e, seconds_e = self.get_hms(self.end_time)
            self.get_batch_data(hours+minutes+seconds, hours_e+minutes_e+seconds_e)
            #self.get_nex_batch()

        ##
        time_key = hours + minutes + seconds
        finish_flg = False
        if len(self.data_dic) == 0:
            if int(hours) >= 15:
                finish_flg = True
            return [], finish_flg
        for symbol in self.data_dic:
            if time_key in self.data_dic[symbol]:
                res.append(self.data_dic[symbol][time_key])
            else:
                if time_key_1 in self.data_dic[symbol]:
                    res.append(self.data_dic[symbol][time_key_1])
                else:
                    if time_key_2 in self.data_dic[symbol]:
                        res.append(self.data_dic[symbol][time_key_2])
        self.tick_i += 1
        return res, finish_flg

    def get_tick_auction(self): # 获取集合竞价tick
        res = None
        self.get_batch_data('92500', '92600')
        for symbol in self.data_dic:
            for time_key in self.data_dic[symbol]:
                res.append(self.data_dic[symbol][time_key])
                break
        return res



    def get_hms(self, sec):
        hours, remainder = divmod(sec, 3600)
        minutes, seconds = divmod(remainder, 60)
        if minutes<10:
            minutes = str(0)+str(minutes)
        if seconds<10:
            seconds = str(0)+str(seconds)
        return str(hours),str(minutes),str(seconds)

def main(HQ_DATA_FILE_TEST, SIMULATE, symbol_test,FEED_TICK_NUM = 500):
    print(' redis mode !!! main ')
    myredis = redis.StrictRedis(decode_responses=True, host="127.0.0.1", port=6379)
    myredis.client_list()
    data_dic = {}
    msqliteObj_feed = sqliteObj_feed(HQ_DATA_FILE_TEST)
    print('db file :',HQ_DATA_FILE_TEST )
    if SIMULATE:
        table_names = msqliteObj_feed.get_table_names()
        __mlistener__ = myredis.pubsub()  # Redis Pubsub 命令用于查看订阅与发布系统状态，它由数个不同格式的子命令组成
        __mlistener__.subscribe([STOP_SEND_CHANNEL])  ##写入数据库监听器1
        batch_num = math.ceil(len(table_names) / 100)
        tick_num = FEED_TICK_NUM
        print('begin send, table_names len:', len(table_names),'batch_num:',batch_num)
        for i in range(tick_num):
            for rept in range(REP_TIME): # 模拟实际情况重发2次
                for j in range(batch_num):
                    data = []
                    for k in range(100):
                        if j * 100 + k >= len(table_names):
                            continue
                        tb='sh'+table_names[j*100+k][0][2:] # 固定sh开头,遗留bug
                        symbol = table_names[j*100+k][0][2:]
                        if symbol not in data_dic:
                            data_all = msqliteObj_feed.fetch_data_from_stock_table(tb,FEED_TICK_NUM+100)
                            # 剔除 9.30分之前
                            data_lst = []
                            for item in data_all[0:tick_num+100]:
                                try:        # '还需要优化有些股票3秒没数据,需要对齐
                                    if int(getContents(item, symbol)[0]['time'])>=93000:
                                        data_lst.append(item)
                                    else:
                                        continue
                                except Exception as e:
                                    print(e)
                            data_dic[symbol]= data_lst
                        else:
                            data_lst = data_dic[symbol]
                        if len(data_lst)>i:
                            data.append(getContents(data_lst[i], symbol)[0])
                    print('len data:', len(data),'tick_i:',i)
                    myredis.publish(const.L2_HQ_CHANNEL_BACK, data)
                    parsed_msg = __mlistener__.get_message()
                    if parsed_msg is not None and parsed_msg != [] \
                            and (parsed_msg['channel'] == STOP_SEND_CHANNEL):
                        data = parsed_msg['data']
                        try:
                            if data != None and type(data) == str:
                                #
                                print('STOP_SEND_CHANNEL data:', data=='1')
                                if data == '1':
                                    msqliteObj_feed.close_db()
                        except:
                            pass
                    if i>0:
                        time.sleep(0.01)
            # print('tick:',i,'time:',time.localtime(time.time()).tm_min,time.localtime(time.time()).tm_sec)


    else:
        data_all=msqliteObj_feed.readdata(symbol_test)
        for i in range(10, len(data_all)):
            data=getContents(data_all[i])
            myredis.publish(const.L2_HQ_CHANNEL, data)
            if i % 500 == 0:
                print('feed tick:',i ,'time:',data[0]['time'])
            time.sleep(0.5)



def getContents(line_detail,symbol):
    resData = []
    try:
            itemTmp = {}
            itemTmp['symbol'] = symbol
            itemTmp['hyd'] = line_detail[0]
            itemTmp['time'] = str(line_detail[1])
            itemTmp['last_close'] = line_detail[2]
            itemTmp['open'] = line_detail[3]
            itemTmp['high'] = line_detail[4]
            itemTmp['low'] = line_detail[5]
            itemTmp['now'] = line_detail[6]
            itemTmp['transaction_count'] = line_detail[7]
            itemTmp['total_volume'] = line_detail[8]
            itemTmp['total_amount'] = line_detail[9]
            itemTmp['current_ask_amount'] = line_detail[10]
            itemTmp['current_bid_amount'] = line_detail[11]
            itemTmp['avrg_bid_price'] = line_detail[12]   #买均
            itemTmp['avrg_ask_price'] = line_detail[13]  #卖均
            #print('itemTmp:',itemTmp)
            resData.append(itemTmp)
    except Exception as e:
        print(e)
    return resData

#  用64位python 运行, 32位内存太小,数据库太大会报错

def delete_data_frm_db(source_folder, save_num=500):   #删除数据库数据,只保留500条记录, 谨慎使用, 做好备份~~~
    for file_db in os.listdir(source_folder):
        msqliteObj_feed = sqliteObj_feed(source_folder + os.sep + file_db)
        table_names = msqliteObj_feed.get_table_names()
        print('file_db names:', file_db)
        for i in range(len(table_names)):
            tb = 'sh' + table_names[i][0][2:]  # 固定sh开头,遗留bug
            sql = 'delete from ' + tb + """ where((time<93000)or(time>95500))"""
            msqliteObj_feed._cursor.execute(sql)

        msqliteObj_feed.conn.commit()
        msqliteObj_feed._cursor.execute('VACUUM')
        msqliteObj_feed.close_db()
#
# if __name__ == "__main__":
#     delete_data_frm_db('E:\\hq_data1')